<?php
$_COOKIE['ver'] = 'online3';
include __DIR__ . '/bootstrap.php';
$zookeeper = \timandes\ZookeeperClient::getInstence();

$result = cache('zookeeper_status');
// 每十分钟更新一次
if ($result && $result['ctime'] > time() - 290) {
    p($result);
    exit;
}
$result = array();
// 读取 zookeeper 数据
$cdate = date("Y-m-d H:i:s");
$data = array("ctime"=>time(),"cdate"=>$cdate);
$path="/storm_kafka";
$data['storm'] = fetch_recursive($path);
$path = "/kafka/consumers/10.207.6.235";
$data['kafka'] = fetch_recursive($path);
$path = "/kafka/consumers/10.207.6.252";
$tmp = fetch_recursive($path);
foreach ($tmp as $k => $v) {
    $data['kafka'][$k] = array_merge($data['kafka'][$k], $v);
}
// 整合 zookeeper 数据
foreach ($data['storm'] as $consumer => $list) {
    if (strstr($consumer, "test") !==false) {
        break;
    }
    $row = array();
    foreach ($list as $item) {
        $item = json_decode($item, true);
        $row || $row = array(
            "cdate" => $data['cdate'],
            "topic" => $item["topic"],
            "storm_topology_id" => $item["topology"]["id"],
            "storm_topology_name" => $item["topology"]["name"],
        );
        $row = array_merge($row,array(
            "storm_offset_".$item["partition"] => $item["offset"],
            "storm_broker_host_".$item["partition"] => $item["broker"]["host"],
            "storm_broker_port_".$item["partition"] => $item["broker"]["port"],
        ));
    }
    $topic = getTopicName($item['topic']);
    $item['topic'] = $topic;
    isset($result[$topic]) || $result[$topic] = array();
    $result[$topic] = array_merge($result[$topic], $row);
}

foreach ($data['kafka']['offsets'] as $topic => $list) {
    $row = array("cdate" => $data['cdate'], 'topic' => $topic);
    foreach ($list as $partition => $offset) {
        if(isset($data['kafka']['owners'][$partition])) {
            $row['flume_owner_'.$partition] = $data['kafka']['offsets']['owners'][$partition];
        }    
        $row['flume_offset_'.$partition] = $offset;
    }
    $topic = getTopicName($topic);
    $row['topic'] = $topic;
    isset($result[$topic]) || $result[$topic] = array();
    $result[$topic] = array_merge($result[$topic], $row);
}
// 读取 elasticsearch 数据
$data = fetch_es_size();
$today = date("Y-m-d");
foreach ($data as $key => $item) {
    if ($item['cdate'] == $today) {
        $item['cdate'] = $cdate;
        $topic = getTopicName($item['topic']);
        $item['topic'] = $topic;
        isset($result[$topic]) || $result[$topic] = array();
        $result[$topic] = array_merge($result[$topic], $item);
    }
}

foreach ($result as & $item) {
    $item["ctime"] = str_replace(" ", "T", $item["cdate"]);
}
$res = save_es($result);
$cache = array("ctime"=>time(),"cdate"=>$cdate, "data" => $result);
cache('zookeeper_status', $cache);

p($result, $res);


function getTopicName($topic) {
    return str_replace(array("_log","_res"), "", $topic);
}

